[Python] Redis 캐싱으로 API 성능 향상시켜보기

JaehyoJJAng2025년 12월 11일

Redis란 무엇인가?

Redis가 왜 필요할까?

상황을 가정해봅시다.

데이터베이스 쿼리: 100ms 소요
하루 API 요청: 10,000회
총 대기 시간: 1,000초 => 16분

이 때 만약 Redis를 도입한다면?

첫 요청: 100ms (DB 조회)
이후 요청: 1ms (캐시 조회)
총 대기 시간: 약 10초

Redis의 핵심 개념

In-Memory: 메모리에 저장 (매우 빠름)
Key-Value: 키로 값을 찾음
TTL: 자동 만료 시간 설정 가능

Redis 동작 방식

1. API 요청 들어옴
2. Redis에 데이터가 있나?
- Yes: 즉시 반환 (1ms) ⚡
- No: DB 조회 (100ms) -> Redis에 저장 -> 반환
3. 다음 요청부터는 Redis에서 즉시 반환 ⚡

1. 실습 환경 구축

Redis 실행 (Docker)

Redis는 도커로 띄우도록 하겠습니다.

docker run -d -it --name redis -p 6379:6379 redis:7-alpine

Redis 실행 테스트

# Redis CLI 접속
docker exec -it redis redis-cli

# 테스트
127.0.0.1:6379> SET test "Hello Redis"
OK

127.0.0.1:6379> GET test
"Hello Redis"

127.0.0.1:6379> exit

파이썬 패키지 설치

실습에 필요한 패키지들을 설치해줍시다.

pip install flask redis mysql-connector-python

flask: 웹 API 서버
redis: Redis 클라이언트
mysql-connector-python: MySQL 연결

2. 데이터베이스 준비

다음과 같은 시나리오를 기반으로 데이터베이스를 구축해보도록 하겠습니다.

대규모 상품 조회 시스템
- 현재 DB에 상품 100,000개 데이터가 존재함.
- 각 상품의 상세 정보 조회 API가 구현되어 있음.
- 매일 수천 건의 API가 요청됨.

DB 스키마 생성

-- redis_practice_db.sql
CREATE DATABASE IF NOT EXISTS redis_practice;
USE redis_practice;

-- 상품 테이블
CREATE TABLE products (
    id INT PRIMARY KEY AUTO_INCREMENT,
    product_code VARCHAR(50) UNIQUE NOT NULL,
    name VARCHAR(200) NOT NULL,
    category VARCHAR(100),
    price INT NOT NULL,
    stock INT DEFAULT 0,
    description TEXT,
    manufacturer VARCHAR(100),
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

    INDEX idx_product_code (product_code),
    INDEX idx_category (category),
    INDEX idx_price (price)
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci ENGINE=InnoDB COMMENT='상품 정보';

-- 상품 상세 정보 테이블 (JOIN 시뮬레이션)
CREATE TABLE product_details (
    id INT PRIMARY KEY AUTO_INCREMENT,
    product_code VARCHAR(50) NOT NULL,
    spec_key VARCHAR(100),
    spec_value TEXT,

    INDEX idx_product_code (product_code),
    FOREIGN KEY (product_code) REFERENCES products(product_code)
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci ENGINE=InnoDB COMMENT='상품 상세 스펙';

-- 재고 이력 테이블 (복잡한 쿼리 시뮬레이션)
CREATE TABLE stock_history (
    id INT PRIMARY KEY AUTO_INCREMENT,
    product_code VARCHAR(50) NOT NULL,
    change_amount INT NOT NULL,
    change_type VARCHAR(20),
    change_date DATETIME DEFAULT CURRENT_TIMESTAMP,

    INDEX idx_product_code (product_code),
    INDEX idx_change_date (change_date),
    FOREIGN KEY (product_code) REFERENCES products(product_code)
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci ENGINE=InnoDB COMMENT='재고 변동 이력';

대용량 샘플 데이터 생성

성능 비교를 위한 대용량 데이터를 한 번 넣어보도록 하겠습니다.

-- sample_data.sql
USE redis_practice;

-- 상품 데이터 (1000개)
DELIMITER //
CREATE PROCEDURE insert_products()
BEGIN
    DECLARE i INT DEFAULT 1;
    DECLARE categories VARCHAR(100);
    DECLARE manufacturers VARCHAR(100);

    WHILE i <= 1000 DO
        SET categories = ELT(FLOOR(1 + RAND() * 5),
            'Electronics', 'Clothing', 'Food', 'Books', 'Toys');
        SET manufacturers = ELT(FLOOR(1 + RAND() * 5),
            'Samsung', 'LG', 'Apple', 'Sony', 'Panasonic');

        INSERT INTO products (product_code, name, category, price, stock, description, manufacturer)
        VALUES (
            CONCAT('PROD-', LPAD(i, 6, '0')),
            CONCAT('Product ', i),
            categories,
            FLOOR(10000 + RAND() * 990000),
            FLOOR(RAND() * 1000),
            CONCAT('This is a detailed description for product ', i, '. High quality and best price!'),
            manufacturers
        );

        SET i = i + 1;
    END WHILE;
END //
DELIMITER ;

-- 프로시저 실행
CALL insert_products();

-- 상품 상세 정보 (각 상품당 3-5개)
INSERT INTO product_details (product_code, spec_key, spec_value)
SELECT
    product_code,
    'Weight',
    CONCAT(FLOOR(100 + RAND() * 9900), 'g')
FROM products;

INSERT INTO product_details (product_code, spec_key, spec_value)
SELECT
    product_code,
    'Dimensions',
    CONCAT(FLOOR(10 + RAND() * 90), 'x', FLOOR(10 + RAND() * 90), 'x', FLOOR(5 + RAND() * 45), 'cm')
FROM products;

INSERT INTO product_details (product_code, spec_key, spec_value)
SELECT
    product_code,
    'Color',
    ELT(FLOOR(1 + RAND() * 5), 'Black', 'White', 'Red', 'Blue', 'Silver')
FROM products;

-- 재고 이력 (각 상품당 10-20건)
INSERT INTO stock_history (product_code, change_amount, change_type)
SELECT
    p.product_code,
    FLOOR(-50 + RAND() * 100),
    ELT(FLOOR(1 + RAND() * 3), 'Purchase', 'Sale', 'Adjustment')
FROM products p
CROSS JOIN (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5) t1
CROSS JOIN (SELECT 1 UNION SELECT 2 UNION SELECT 3) t2;

-- 데이터 확인
SELECT COUNT(*) as product_count FROM products;
SELECT COUNT(*) as detail_count FROM product_details;
SELECT COUNT(*) as history_count FROM stock_history;

데이터를 DB에 넣어줄게요.

mysql -u root -p < redis_pratice_db.sql
mysql -u root -p < sample_data.sql

Flask API 구현

config.py - 설정 파일

DB 접근 및 redis 접근을 위한 Config 클래스를 생성해주도록 하겠습니다.

class Config():
    # MySQL 설정
    MYSQL_HOST: str = 'localhost'
    MYSQL_PORT: int = 3306
    MYSQL_USER: str = 'root'
    MYSQL_PASSWORD: str = 'custom'
    MYSQL_DATABASE: str = 'redis_practice'

    # Redis 설정
    REDIS_HOST: str = 'localhost'
    REDIS_PORT: int = 6379
    REDIS_DB: int = 0
    REDIS_DECODE_RESPONSES: bool = True

    # 캐시 TTL (초)
    CACHE_TTL: int = 3600 # 1시간

db.py - DB 연결

import mysql.connector as conn
from config import Config


def get_db_connection() -> conn.connection_cext.CMySQLConnection:
    return conn.connect(
        host=Config.MYSQL_HOST,
        port=Config.MYSQL_PORT,
        user=Config.MYSQL_USER,
        password=Config.MYSQL_PASSWORD,
        database=Config.MYSQL_DATABASE,
    )


def get_product_detail(product_code: str) -> dict:
    """
    상품 상세 정보 조회 (복잡한 쿼리)
    - products 테이블
    - product_details 조인
    - stock_history 집계
    """
    with get_db_connection() as conn:
        cursor = conn.cursor(dictionary=True)

        # 복잡한 쿼리 (의도적으로 느리게)
        query = """
            SELECT
                p.product_code,
                p.name,
                p.category,
                p.price,
                p.stock,
                p.description,
                p.manufacturer,
                GROUP_CONCAT(
                    DISTINCT CONCAT(pd.spec_key, ':', pd.spec_value)
                    SEPARATOR '||'
                ) as specifications,
                COUNT(DISTINCT sh.id) as stock_changes,
                SUM(CASE WHEN sh.change_type = 'Sale' THEN sh.change_amount ELSE 0 END) as total_sales
            FROM products p
            LEFT JOIN product_details pd ON p.product_code = pd.product_code
            LEFT JOIN stock_history sh ON p.product_code = sh.product_code
            WHERE p.product_code = %s
            GROUP BY p.product_code, p.name, p.category, p.price,
                    p.stock, p.description, p.manufacturer
        """

        cursor.execute(query, (product_code,))
        result: dict = cursor.fetchone()
    return result


def get_products_by_category(category: str, limit=10) -> list[dict]:
    "카테고리별 상품 목록 조회"
    with get_db_connection() as conn:
        cursor = conn.cursor(dictionary=True)
        query = """
            SELECT
                p.product_code,
                p.name,
                p.category,
                p.price,
                p.stock,
                p.manufacturer,
                COUNT(sh.id) as total_transactions
            FROM products p
            LEFT JOIN stock_history sh ON p.product_code = sh.product_code
            WHERE p.category = %s
            GROUP BY p.product_code, p.name, p.category, p.price, p.stock, p.manufacturer
            ORDER BY p.price DESC
            LIMIT %s
        """
        cursor.execute(
            query,
            (
                category,
                limit,
            ),
        )
        result: list[dict] = cursor.fetchall()
    return result

app.py - Flask API (No REDIS)

Redis를 적용하지 않은 API 코드입니다.

from flask import Flask, jsonify, request, Response
from db import get_db_connection, get_products_by_category, get_product_detail
import time


app = Flask(__name__)


@app.route("/api/product/<product_code>")
def api_product_detail(product_code: str) -> Response:
    start_time = time.time()

    # DB 조회
    product: dict = get_product_detail(product_code=product_code)

    if not product:
        return jsonify({"error": "product not found"}), 404

    # 스펙 파싱
    try:
        if product.get("specifications"):
            specs = {}
            for spec in product.get("specifications").split("||"):
                key, value = spec.split(":")
                specs[key] = value
            product["specifications"] = specs

        elapsed_time = (time.time() - start_time) * 1000  # ms
        return jsonify(
            {
                "success": True,
                "data": product,
                "query_time_ms": round(elapsed_time, 2),
                "cached": False,
            }
        )
    except Exception as e:
        return jsonify({"success": False, "message": str(e)})


@app.route("/api/products/category/<category>")
def api_products_by_category(category: str) -> Response:
    """카테고리별 상품 목록 API"""
    start_time = time.time()

    limit = request.args.get("limit", 10, type=int)
    products: list[dict] = get_products_by_category(category=category, limit=limit)
    if not products:
        return (
            jsonify(
                {"success": False, "message": "가져올 수 있는 상품 목록이 없습니다."}
            ),
            400,
        )
    elapsed_time = (time.time() - start_time) * 1000

    return jsonify(
        {
            "success": True,
            "data": products,
            "count": len(products),
            "query_time_ms": round(elapsed_time, 2),
            "cached": False,
        }
    )


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

테스트 실행

# Flask 서버 실행
python app.py

# 다른 터미널에서 테스트
curl http://localhost:5000/api/product/PROD-000001
curl http://localhost:5000/api/products/category/Electronics

# 결과
{
  "success": true,
  "data": {
    "product_code": "PROD-000001",
    "name": "Product 1",
    "price": 450000,
    "stock": 523
  },
  "query_time_ms": 125.34,
  "cached": false
}

현재 API 조회 시 100ms 이상 소요되는 것을 볼 수 있습니다.

Redis 캐싱 적용

`app.py` - Redis 적용

기존에 작성했던 app.py 파일에 Redis를 적용하는 코드를 추가하겠습니다.

# app_with_redis.py
from flask import Flask, jsonify, request
import redis
import json
import time
from config import Config
from db import get_product_detail, get_products_by_category

app = Flask(__name__)

# Redis 클라이언트 초기화
redis_client = redis.Redis(
    host=Config.REDIS_HOST,
    port=Config.REDIS_PORT,
    db=Config.REDIS_DB,
    decode_responses=Config.REDIS_DECODE_RESPONSES
)

def get_cached_or_query(cache_key, query_func, *args, ttl=Config.CACHE_TTL):
    """
    캐시 조회 또는 DB 쿼리

    동작 순서:
    1. Redis에서 캐시 확인
    2. 있으면 즉시 반환 (CACHE HIT)
    3. 없으면 DB 조회 → Redis 저장 → 반환 (CACHE MISS)
    """
    # 1. 캐시 확인
    cached_data = redis_client.get(cache_key)

    if cached_data:
        # CACHE HIT ⚡
        return json.loads(cached_data), True

    # 2. CACHE MISS - DB 조회
    data = query_func(*args)

    if data:
        # 3. Redis에 저장
        redis_client.setex(
            cache_key,
            ttl,
            json.dumps(data, ensure_ascii=False, default=str)
        )

    return data, False

@app.route('/api/product/<product_code>')
def api_product_detail(product_code):
    """상품 상세 정보 API (Redis 캐싱 적용)"""
    start_time = time.time()

    # 캐시 키 생성
    cache_key = f"product:{product_code}"

    # 캐시 조회 또는 DB 쿼리
    product, is_cached = get_cached_or_query(
        cache_key,
        get_product_detail,
        product_code
    )

    if not product:
        return jsonify({'error': 'Product not found'}), 404

    # 스펙 파싱
    if product.get('specifications'):
        specs = {}
        for spec in product['specifications'].split('||'):
            if ':' in spec:
                key, value = spec.split(':', 1)
                specs[key] = value
        product['specifications'] = specs

    elapsed_time = (time.time() - start_time) * 1000

    return jsonify({
        'success': True,
        'data': product,
        'query_time_ms': round(elapsed_time, 2),
        'cached': is_cached,
        'cache_key': cache_key if is_cached else None
    })

@app.route('/api/products/category/<category>')
def api_products_by_category(category):
    """카테고리별 상품 목록 API (Redis 캐싱 적용)"""
    start_time = time.time()

    limit = request.args.get('limit', 10, type=int)
    cache_key = f"category:{category}:limit:{limit}"

    # 캐시 조회 또는 DB 쿼리
    products, is_cached = get_cached_or_query(
        cache_key,
        get_products_by_category,
        category,
        limit
    )

    elapsed_time = (time.time() - start_time) * 1000

    return jsonify({
        'success': True,
        'data': products,
        'count': len(products),
        'query_time_ms': round(elapsed_time, 2),
        'cached': is_cached
    })

@app.route('/api/cache/clear')
def clear_cache():
    """캐시 전체 삭제"""
    redis_client.flushdb()
    return jsonify({'success': True, 'message': 'Cache cleared'})

@app.route('/api/cache/stats')
def cache_stats():
    """캐시 통계"""
    info = redis_client.info('stats')
    return jsonify({
        'total_keys': redis_client.dbsize(),
        'hits': info.get('keyspace_hits', 0),
        'misses': info.get('keyspace_misses', 0),
        'hit_rate': round(
            info.get('keyspace_hits', 0) /
            max(info.get('keyspace_hits', 0) + info.get('keyspace_misses', 0), 1) * 100,
            2
        )
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

성능 비교

먼저 Redis를 적용하지 않고 /api/products/PROD-001000 엔드포인트 호출 시 다음과 같은 정보가 리턴됩니다.

{'cached': False, 'data': {'category': 'Electronics', 'description': 'This is a detailed description for product 1000. High quality and best price!', 'manufacturer': 'Sony', 'name': 'Product 1000', 'price': 138147, 'product_code': 'PROD-001000', 'specifications': {'Color': 'White', 'Dimensions': '26x19x48cm', 'Weight': '8076g'}, 'stock': 537, 'stock_changes': 15, 'total_sales': '-348'}, 'query_time_ms': 30.03, 'success': True}

query_time_ms가 30.03로 측정되네요.

이번에는 Redis를 적용하여 /api/products/PROD_001000 엔드포인트를 호출해보겠습니다.

{'cached': False, 'data': {'category': 'Electronics', 'description': 'This is a detailed description for product 1000. High quality and best price!', 'manufacturer': 'Sony', 'name': 'Product 1000', 'price': 138147, 'product_code': 'PROD-001000', 'specifications': {'Color': 'White', 'Dimensions': '26x19x48cm', 'Weight': '8076g'}, 'stock': 537, 'stock_changes': 15, 'total_sales': '-348'}, 'query_time_ms': 0.41, 'success': True}

query_time_ms가 0.41로 호출 시간이 대폭 줄어든 것을 볼 수 있습니다.

마무리

1. Redis 핵심 개념

# Key-Value 저장
redis_client.set('key', 'value')
redis_client.get('key')

# TTL 설정 (자동 만료)
redis_client.setex('key', 3600, 'value')  # 1시간 후 삭제

# 캐시 패턴
cached = redis_client.get(cache_key)
if cached:
    return cached  # HIT
else:
    data = query_db()  # MISS
    redis_client.setex(cache_key, ttl, data)
    return data

2. 캐시 키 설계

# 좋은 예
f"product:{product_code}"
f"category:{category}:limit:{limit}"
f"user:{user_id}:profile"

# 나쁜 예
f"data"  # 너무 일반적
f"{product_code}"  # 의미 불명확

redis